package com.scala.ExcelDataInput

import java.io.InputStream

import com.java.config.IpConfig
import com.java.util.PatternUtil
import com.scala.CopeWithData.{CopeWithDataForNearBy, CopeWithDataForRentInfo}
import com.scala.conf.javaExcel
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Component


/**
 * @Author: ZhangJin
 * @Date: 2020/9/20 22:04
 */

@Component(value = "DataInputFromExcel")
class DataInputFromExcel @Autowired()(val session: SparkSession,val javaExcel:javaExcel,val ipconfig:IpConfig,copewithDataNearBy: CopeWithDataForNearBy,copeWithDataForRentInfo: CopeWithDataForRentInfo) extends Serializable {


  def getDataFromExcel(inputStream:InputStream
  ,tableName: String, SQLforTable: String): Unit = {
    /* 生成随机的字符串作为表名 */
    val UUidString: String = PatternUtil.RandomString
    javaExcel.readExcel(inputStream, UUidString)
    val frame: DataFrame = session.sql("select * from global_temp." + UUidString)
    /* 全局视图进行持久化 */
    frame.write.mode(SaveMode.Overwrite).parquet("hdfs://"+ ipconfig.getIp+ ":" +ipconfig.getPort + "/usr/data/" + tableName + ".parquet")
    session.sqlContext.sql("use default")
    session.sqlContext.sql("create table if not exists "  + tableName + SQLforTable + " stored as parquet")
    session.sqlContext.sql("load data inpath 'hdfs://" + ipconfig.getIp+ ":" +ipconfig.getPort + "/usr/data/" + tableName + ".parquet' overwrite into table " + tableName)
    /* 用完删除临时全局视图  防止占空间 */
    session.catalog.dropGlobalTempView(UUidString)
    // TODO: 数据处理
    copeWithDataForRentInfo.copeWithDataForRent(tableName,SQLforTable)
  }


  def getDataFromExcelOfNearBy(inputStream:InputStream
                       ,tableName: String, SQLforTable: String): Unit = {
    /* 生成随机的字符串作为表名 */
    val UUidString: String = PatternUtil.RandomString
    javaExcel.readExcel(inputStream, UUidString)
    val frame: DataFrame = session.sql("select * from global_temp." + UUidString)
    /* 全局视图进行持久化 */
    val hc=session.sqlContext
    hc.sql("use default")
    frame.write.mode(SaveMode.Overwrite).parquet("hdfs://"+ ipconfig.getIp+ ":" +ipconfig.getPort + "/usr/data/" + tableName + ".parquet")
    hc.sql("create table if not exists " + tableName + SQLforTable + " stored as parquet")
    hc.sql("load data inpath 'hdfs://" + ipconfig.getIp+ ":" +ipconfig.getPort + "/usr/data/" + tableName + ".parquet' overwrite into table " + tableName)
    /* 用完删除临时全局视图  防止占空间 */
    session.catalog.dropGlobalTempView(UUidString)
    // TODO: 数据处理
   copewithDataNearBy.CopeNearByData(tableName,SQLforTable)
  }

}
